package com.yjjxt

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * 求出总售价额TOP3的品牌
 */
object Hello25TopN {
  def main(args: Array[String]): Unit = {
    //创建SparkContext
    val sparkContext = new SparkContext((new SparkConf().setMaster("local").setAppName("Hello25TopN" + System.currentTimeMillis())))
    //开始读取数据
    val lines: RDD[String] = sparkContext.textFile("src/main/resources/topN.txt", 5)
    //开始处理数据（总售价额TOP3的品牌）
    //    val pp: RDD[(String, Int)] = lines.map(ele => (ele.split("\\s")(0), ele.split("\\s")(2).toInt)).reduceByKey(_ + _)
    //    pp.sortBy(_._2, false).take(3).foreach(println)
    //开始处理数据（每种品牌TOP3的种类）
    //    val pp = lines.map(ele => {
    //      val strs = ele.split("\\s")
    //      ((strs(0), strs(1)), strs(2).toInt)
    //    }).reduceByKey(_ + _).sortBy(_._2, false).groupBy(_._1._1).foreach(ele => {
    //      println(ele._1 + "==>" + ele._2.take(3))
    //    })

    //开始处理数据（每种品牌TOP3的种类）

    //首先求出每种品类的总数
    /**
     * (阿迪，裤子) 100
     * (阿迪，袜子) 20
     */


    val pp1: RDD[((String, String), Int)] = lines.map(ele => {
      val strs = ele.split("\\s")
      ((strs(0), strs(1)), strs(2).toInt)
    })

    val pp2: RDD[((String, String), Int)] = lines.map(ele => {
      val strs = ele.split("\\s")
      ((strs(0), strs(1)), 1)
    })


    val ppSum: RDD[((String, String), Int)] = pp1.reduceByKey(_ + _)
    val ppCount: RDD[((String, String), Int)] = pp2.reduceByKey(_ + _)

    ppSum.join(ppCount).map(ele => {
      (ele._1, ele._2._1 / ele._2._2)
    }).sortBy(_._2, false).groupBy(_._1._1).foreach(ele => {
      println(ele)
    })
  }
}
